package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.ExamGardeTestBean;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.util.ClickHouseUtil;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import com.atguigu.edu.realtime.util.TimestampLtz3CompareUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.TimeUnit;


public class DwsExamGradeTestWindow {



    public static void main(String[] args) throws Exception {

        // TODO 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment
                .getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2. 检查点相关设置(略)
        // TODO 3. 从 Kafka dwd_trade_order_detail 主题读取订单明细数据
        String topic = "dwd_interaction_exam";
        String groupId = "dws_exam_paper_grade_test_window";

        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> source = env.addSource(kafkaConsumer);

        SingleOutputStreamOperator<String> filterDS = source.filter(
                new FilterFunction<String>() {
                    @Override
                    public boolean filter(String jsonStr) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);

                        String paperId = jsonObj.getString("paper_id");


                        return paperId != null;


                    }
                }
        );

        // TODO 4.转换数据结构
        SingleOutputStreamOperator<JSONObject> mappedStream = filterDS.map(JSON::parseObject);

        // TODO 5. 按照 order_detail_id 分组
        KeyedStream<JSONObject, String> keyedStream = mappedStream.keyBy(r -> r.getString("id"));
        SingleOutputStreamOperator<JSONObject> processedStream = keyedStream.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    private ValueState<JSONObject> lastValueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        lastValueState = getRuntimeContext().getState(
                                new ValueStateDescriptor<JSONObject>("last_value_state", JSONObject.class)
                        );
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject lastValue = lastValueState.value();
                        if (lastValue == null) {
                            long currentProcessingTime = ctx.timerService().currentProcessingTime();
                            ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                            lastValueState.update(jsonObj);
                        } else {
                            String lastRowOpTs = lastValue.getString("row_op_ts");
                            String rowOpTs = jsonObj.getString("row_op_ts");
                            if (TimestampLtz3CompareUtil.compare(lastRowOpTs, rowOpTs) <= 0) {
                                lastValueState.update(jsonObj);
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws IOException {
                        JSONObject lastValue = this.lastValueState.value();
                        if (lastValue != null) {
                            out.collect(lastValue);
                        }
                        lastValueState.clear();
                    }
                }
        );



        // TODO 7. 转换数据结构
        SingleOutputStreamOperator<ExamGardeTestBean> javaBeanStream = processedStream.map(
                jsonObj -> {
                    String paperId = jsonObj.getString("paper_id");
                    String userId = jsonObj.getString("user_id");
                    Double score = jsonObj.getDouble("score");
                    Long ts = jsonObj.getLong("ts") * 1000L;

                    ExamGardeTestBean examGardeTestBean = ExamGardeTestBean.builder()
                            .paperId(paperId)
                            .userId(userId)
                            .score(score)
                            .ts(ts)
                            .build();
                    return examGardeTestBean;
                }
        );



        // TODO 9. 按照省份 ID 分组
        KeyedStream<ExamGardeTestBean, String> keyedByProIdStream =
                javaBeanStream.keyBy(ExamGardeTestBean::getPaperId);

        SingleOutputStreamOperator<ExamGardeTestBean> uuCountDS = keyedByProIdStream.process(
                new KeyedProcessFunction<String, ExamGardeTestBean, ExamGardeTestBean>() {

                    private MapState<String,String> lastOrderDt;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        MapStateDescriptor<String,String> mapStateDescriptor =
                                new MapStateDescriptor<>("grade_window_last_order_dt", String.class,String.class);
                        // 设置状态 ttl
                        mapStateDescriptor.enableTimeToLive(
                                StateTtlConfig
                                        .newBuilder(Time.days(1L))
                                        .updateTtlOnCreateAndWrite()
                                        .build()
                        );
                        lastOrderDt = getRuntimeContext().getMapState(mapStateDescriptor);
                    }

                    @Override
                    public void processElement(ExamGardeTestBean value, KeyedProcessFunction<String, ExamGardeTestBean, ExamGardeTestBean>.Context ctx, Collector<ExamGardeTestBean> out) throws Exception {
                        Long ts = value.getTs();
                        String curDate = DateFormatUtil.toDate(ts);
                        String userId = value.getUserId();
                        String lastDate = lastOrderDt.get(userId);
                        Double score = value.getScore();
                        if (lastDate == null || lastDate == curDate) {
                            if(score < 60.0){
                                value.setFail(1L);
                                value.setGood(0L);
                                value.setExcellent(0L);
                            }
                            else if (score >= 60.0 && score <80.0) {
                                value.setFail(0L);
                                value.setGood(1L);
                                value.setExcellent(0L);
                            }
                            else {
                                value.setFail(0L);
                                value.setGood(0L);
                                value.setExcellent(1L);
                            }
                            lastOrderDt.put(userId,lastDate);
                        }
                        out.collect(value);

                    }


                }
        );

        // TODO 10. 设置水位线
        SingleOutputStreamOperator<ExamGardeTestBean> watermarkDS = uuCountDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<ExamGardeTestBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<ExamGardeTestBean>() {
                                    @Override
                                    public long extractTimestamp(ExamGardeTestBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )
        );

        KeyedStream<ExamGardeTestBean, String> keyedForAggregateStream = watermarkDS.keyBy(
                new KeySelector<ExamGardeTestBean, String>() {

                    @Override
                    public String getKey(ExamGardeTestBean javaBean) throws Exception {
                        return javaBean.getPaperId();
                    }
                }
        );

        // TODO 10. 开窗
        WindowedStream<ExamGardeTestBean, String, TimeWindow> windowDS = keyedForAggregateStream.window(TumblingEventTimeWindows.of(
                org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)
        ));

        // TODO 11. 聚合计算
        SingleOutputStreamOperator<ExamGardeTestBean> reducedStream = windowDS.reduce(
                new ReduceFunction<ExamGardeTestBean>() {
                    @Override
                    public ExamGardeTestBean reduce(ExamGardeTestBean value1, ExamGardeTestBean value2) throws Exception {
                        value1.setExcellent(
                                value1.getExcellent() + value2.getExcellent()
                        );
                        value1.setGood(
                                value1.getGood()+value2.getGood()
                        );
                        value1.setFail(
                                value1.getFail()+value2.getFail()
                        );
                        return value1;
                    }
                },
                new ProcessWindowFunction<ExamGardeTestBean, ExamGardeTestBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<ExamGardeTestBean> elements, Collector<ExamGardeTestBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        for (ExamGardeTestBean element : elements) {
                            element.setStt(stt);
                            element.setEdt(edt);
                            element.setTs(System.currentTimeMillis());
                            out.collect(element);
                        }
                    }
                }
        );

        SingleOutputStreamOperator<ExamGardeTestBean> fullInfoStream = AsyncDataStream.unorderedWait(
                reducedStream,
                new DimAsyncFunction<ExamGardeTestBean>("dim_test_paper".toUpperCase()) {

                    @Override
                    public void join(ExamGardeTestBean javaBean, JSONObject jsonObj) throws Exception {
                        String paperName = jsonObj.getString("paper_title".toUpperCase());
                        javaBean.setPaperName(paperName);
                    }

                    @Override
                    public String getKey(ExamGardeTestBean javaBean) {
                        return javaBean.getPaperId();
                    }
                },
                60 * 50, TimeUnit.SECONDS
        );




        SinkFunction<ExamGardeTestBean> jdbcSink = ClickHouseUtil.<ExamGardeTestBean>getJdbcSink(
                "insert into dws_exam_paper_grade_test_window values(?,?,?,?,?,?,?,?)"
        );
        fullInfoStream.<ExamGardeTestBean>addSink(jdbcSink);

        fullInfoStream.print(">>>");


        env.execute();
    }

}
